package com.cloud.message.support;

import com.cloud.common.bean.Message;
import com.cloud.common.bean.ResultInfo;
import com.cloud.common.bean.SessionInfo;
import com.cloud.common.context.SessionContext;
import com.cloud.common.context.VersionContext;
import com.cloud.common.rpc.RPCInvoker;
import com.cloud.common.sentinel.Sentinel;
import com.cloud.common.util.JsonUtils;
import com.cloud.message.cache.ConsumerCache;
import com.cloud.message.entity.Consumer;
import com.cloud.message.entity.MessagePending;
import com.cloud.message.rocketmq.MQMessage;
import com.cloud.message.service.MessagePendingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;

@Slf4j
@Sentinel
@Component
public class MessageSender {

    @Autowired
    private ConsumerCache consumerCache;

    @Autowired
    private RPCInvoker rpcInvoker;

    @Autowired
    private MessagePendingService messagePendingService;

    public void sendMessage(MQMessage mqMsg) {
        List<Consumer> consumerList = consumerCache.getConsumerList(mqMsg.getTopic());
        if (CollectionUtils.isEmpty(consumerList)) {
            return;
        }
        VersionContext.set(mqMsg.getUserVersion());
        SessionContext.set(JsonUtils.parseObject(mqMsg.getUserSession(), SessionInfo.class));
        try {
            for (Consumer consumer : consumerList) {
                Message message = new Message();
                BeanUtils.copyProperties(mqMsg, message);
                if (!doSend(consumer.getServiceName(), message)) {
                    messagePendingService.createPending(consumer.getServiceName(), mqMsg);
                }
            }
        } finally {
            VersionContext.remove();
            SessionContext.remove();
        }
    }

    public void sendPending(MessagePending pending) {
        VersionContext.set(pending.getUserVersion());
        SessionContext.set(JsonUtils.parseObject(pending.getUserSession(), SessionInfo.class));
        try {
            Message message = new Message();
            BeanUtils.copyProperties(pending, message);
            if (!doSend(pending.getServiceName(), message)) {
                pending.setRetryCount(pending.getRetryCount() + 1);
                pending.setUpdateTime(new Date());
                messagePendingService.updateById(pending);
            } else {
                messagePendingService.removeById(pending.getId());
            }
        } finally {
            VersionContext.remove();
            SessionContext.remove();
        }
    }

    private boolean doSend(String serviceName, Message message) {
        try {
            String requestUri = "/message/handle";
            String[] parameterNames = new String[]{"message"};
            Object[] parameterValues = new Object[]{message};
            int timeout = 15000;
            ResultInfo resultInfo = (ResultInfo) rpcInvoker.invoke(serviceName, requestUri, timeout, parameterNames, parameterValues, ResultInfo.class);
            return resultInfo.successful();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        log.error("send failure serviceName=" + serviceName);
        return false;
    }

}
